package org.example

import org.apache.spark.sql.SparkSession
import java.util.regex.Pattern


object movie {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext
 //  用户分析
    val filePath ="src/main/resources/"
    val moviesRDD =sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD=sc.textFile(filePath + "ratings.dat")
    val usersRDD =sc.textFile(filePath +"users.dat")
    println("总电影数："+moviesRDD.count())
    println("总用户数："+ usersRDD.count())
    println("评分条数："+ ratingsRDD.count())
    println("职业数:" + occupationsRDD.count())
// 用户分析统计
    val userRating = ratingsRDD.map(_.split("::"))
      .map(user =>(user(1),user(0)))
      .filter(_._2.equals("2688"))
    println("用户id为2688"+ userRating.count())




//
val rating = ratingsRDD.map(x => x.split("::")).map {
  x => {
     (x(0), x(1), x(2))
      }
}.cache()
    println("平均得分最高的前 10 名的电影名称简单版")

rating.map(x => (x._2, (x._3.toDouble, 1)))
  .reduceByKey((x, y) => {
    (x._1 + y._1, x._2 + y._2)
  })
  .map(x => (x._2._1 / x._2._2, x._1))
  .sortByKey(false)
  .take(5)
  .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")


  val moviesInfo = moviesRDD.map(x => x.split("::"))
    .map(x =>{
      (x(0), (x(1), x(2)))
    })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x,y)=>{
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))

    moviesInfo.join(ratingsInfo)
      .map(info =>{
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
      }).sortByKey(false)
      .take(5)
      .foreach(println)
    println("观影人数最多的前 10 部电影")

    val watchViewsInfo = rating.map(x => {
      (x._2, 1)
    }).reduceByKey((x, y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)

    watchViewsInfo.foreach(println(_))
    println("===================>")

    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) =>{
        (x+y)
      })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)

    println("详情的输出(  观影人数，电影编号)")

    moviesInfo.join(ratingsInfo).map(x => {
      (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
    }).sortByKey(false)
      .take(10)
      .foreach(println)

    val movieStyle = moviesRDD.map(_.split("::"))
      .map(tp => (tp(0),tp(2)))
      .flatMapValues(sp => sp.split("\\|"))
      .map(p => (p._2,1))
      .reduceByKey((x,y) => x +y)
      .filter(_._1.equals("Comedy"))

    val pattern = Pattern.compile(" (.*) (\\(\\d{4}\\))")
    val movieInfo = moviesRDD
      .map(x => x.split("::"))
      .map(x =>{
        (x(1),1)
      })
      .map(x => {
        var name =""
        var year =""
        val matcher = pattern.matcher(x._1)
        if (matcher.find()) {
          name = matcher.group(1)
          year = matcher.group(2)
          year = year.substring(1, year.length() - 1)
        }
        if (year == "") {
          (-1,1)
        }else {
          (year.toInt,1)
        }
      })
      .reduceByKey((x,y) => x+y)
      .sortByKey()
      .collect()

    movieInfo.take(10).foreach(println(_))



    sc.stop()
  }
}
